package com.rephilo.hbase;

import org.apache.commons.collections.CollectionUtils;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.CoprocessorEnvironment;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessor;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.coprocessor.RegionObserver;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.wal.WALEdit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;

/**
 * 同步删除协处理器
 *
 * @author rephilo
 */
public class DeleteProcessor implements RegionObserver, RegionCoprocessor {

    private static final Logger logger = LoggerFactory.getLogger(DeleteProcessor.class);

    private String RELATIONS_TABLE_NAME = "relations";

    private Connection conn;

    @Override
    public Optional<RegionObserver> getRegionObserver() {
        return Optional.of(this);
    }

    @Override
    public void start(CoprocessorEnvironment env) throws IOException {
        conn = ConnectionFactory.createConnection(env.getConfiguration());
    }

    @Override
    public void stop(CoprocessorEnvironment env) throws IOException {
        conn.close();
    }

    @Override
    public void postDelete(ObserverContext<RegionCoprocessorEnvironment> e, Delete delete, WALEdit edit) throws IOException {
        TableName tableName = e.getEnvironment().getRegion().getRegionInfo().getTable();
        //判断当前操作是否是relations表
        if (!new String(tableName.getName()).equals(this.RELATIONS_TABLE_NAME)) {
            return;
        }

        //判断操作friends列族是否为空
        List<Cell> cells = delete.getFamilyCellMap().get(Bytes.toBytes("friends"));
        if (CollectionUtils.isEmpty(cells)) {
            return;
        }

        Table table = conn.getTable(tableName);

        for (Cell cell : cells) {
            byte[] column = CellUtil.cloneQualifier(cell);
            byte[] rowKey = CellUtil.cloneRow(cell);

            //同步删除
            if (column.length > 0 && rowKey.length > 0) {
                //查询要删除的记录是否存在
                Get get = new Get(column);
                get.addColumn(Bytes.toBytes("friends"), rowKey);
                Result result = table.get(get);
                if (result.rawCells() != null && result.rawCells().length > 0) {
                    //查询结果存在则同步删除
                    Delete otherUserDelete = new Delete(column);
                    otherUserDelete.addColumns(Bytes.toBytes("friends"), rowKey);
                    table.delete(otherUserDelete);

                    logger.info("delete family:{}, column: {}, value:{}, rowkey: {}", Bytes.toString(CellUtil.cloneFamily(cell)),
                            Bytes.toString(column), Bytes.toString(CellUtil.cloneValue(cell)), Bytes.toString(rowKey));
                }
            }
        }
    }
}
